package cn.spring.rabbitmq.demo.config;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;

@Configuration
@Import(RabbitmqTemplatePostProcessor.class)
public class RabbitMQConfig {

    /**
     * 延迟交换机（用于接收延迟消息）
     *
     * @return
     */
    @Bean
    public Exchange delayExchange() {
        return ExchangeBuilder.directExchange("delay.exchange")
                .durable(true)
                .build();
    }

    /**
     * 延迟队列（没有消费者监听）
     * 设置死信交换机和路由键，当消息过期后自动转发到死信交换机
     *
     * @return
     */
    @Bean
    public Queue delayQueue() {
        return QueueBuilder.durable("delay.queue")
                // 设置死信交换机
                .deadLetterExchange("delay.process.exchange")
                // 设置死信路由键
                .deadLetterRoutingKey("delay.process")
                .build();
    }

    /**
     * 延迟交换机与延迟队列绑定
     *
     * @return
     */
    @Bean
    public Binding delayBinding() {
        return BindingBuilder.bind(delayQueue())
                .to(delayExchange())
                .with("delay")
                .noargs();
    }


    /**
     * json消息转换器
     *
     * @return
     */
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    /**
     * 为批量处理创建专用的监听器容器工厂
     *
     * @param connectionFactory
     * @return
     */
    @Bean("batchListenerContainerFactory")
    public SimpleRabbitListenerContainerFactory batchListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setConcurrentConsumers(3);
        factory.setMaxConcurrentConsumers(3);
        factory.setPrefetchCount(10);
        factory.setBatchListener(true); // 启用批量监听
        factory.setConsumerBatchEnabled(true); // 启用消费者端批量处理
        factory.setBatchSize(10); // 设置批次大小
        factory.setReceiveTimeout(5000L); // 设置接收超时时间
        factory.setBatchReceiveTimeout(5000L);
        factory.setDeBatchingEnabled(false); // 禁用分解批处理消息
        factory.setMessageConverter(messageConverter());
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 设置为 true 可在当前消息处理完毕后停止容器，并重新排队任何预取的消息。在使用独占或单活动消费者时很有用
        factory.setForceStop(true);
        // 设置关闭超时时间
        factory.setContainerCustomizer((container -> {
            container.setShutdownTimeout(30000L);
            // container.setExclusive(true);
        }));
        return factory;
    }
}
